package com.atguigu.sink;

import com.alibaba.fastjson.JSON;
import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Properties;

/**
 * Created by RUI on 2021/6/10 16:35
 */
public class Kafka_Sink {

    public static void main(String[] args) throws Exception {

        ArrayList<WaterSensor> waterSensors = new ArrayList<>();
        waterSensors.add(new WaterSensor("sensor_1", 1607527992000L, 20));
        waterSensors.add(new WaterSensor("sensor_1", 1607527994000L, 50));
        waterSensors.add(new WaterSensor("asensor_1", 1607527996000L, 50));
        waterSensors.add(new WaterSensor("sensor_2", 1607527993000L, 10));
        waterSensors.add(new WaterSensor("sensor_2", 1607527995000L, 30));

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);

        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers","hadoop162:9092");
        env
                .fromCollection(waterSensors)
                .map(JSON::toJSONString)
//                .addSink(new FlinkKafkaProducer<String>("hadoop162:9092","sensor",new SimpleStringSchema()));
                .addSink(new FlinkKafkaProducer<String>(
                        "default",
                        new KafkaSerializationSchema<String>() {
                            @Override
                            public ProducerRecord<byte[], byte[]> serialize(String element, @Nullable Long timestamp) {
                                return new ProducerRecord<byte[], byte[]>("sensor1",JSON.toJSONBytes(element));
                            }
                        },
                        prop,
                        FlinkKafkaProducer.Semantic.EXACTLY_ONCE

                ));

        env.execute();



    }
}
